package com.test.zk;

import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;


import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * Created by xingfeng on 19-3-4.
 */
@Slf4j
public class ZKNodeOperator implements Watcher {
    @Getter
    @Setter
    private ZooKeeper zooKeeper;

    public static final String zkServerPath = "127.0.0.1:2181"; //连接单击
    public static final String zkClusterServerPath = "127.0.0.1:2182,127.0.0.1:2183,,127.0.0.1:2184";//连接集群
    public static final Integer timeout = 5000;

    private static CountDownLatch countDownLatch = new CountDownLatch(1);

    public ZKNodeOperator(){}

    public ZKNodeOperator(String connString){
        try {
            zooKeeper = new ZooKeeper(zkServerPath,timeout,new ZKNodeOperator());
        } catch (IOException e) {
            e.printStackTrace();
            if (null != zooKeeper){
                try {
                    zooKeeper.close();
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }
    }

    @SneakyThrows
    public static void main(String[] args) {
        ZKNodeOperator zkServer = new ZKNodeOperator(zkServerPath);

        //创建节点  任何人都可以访问
//        zkServer.zkNodeCreate("/testnode","testnode".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE);

        //同步修改节点    最后一个版本号可通过get /testnode参看dataverion获得
//        Stat stat = zkServer.getZooKeeper().setData("/testnode","modifytestnode".getBytes(),0);
//        log.warn("同步修改后节点版本："+stat.getVersion());

        //异步修改节点
//        String ctx = "{'update':'success'}";
//        zkServer.getZooKeeper().setData("/testnode","modifytestnode2".getBytes(),1,new UpdateCallback(),ctx);
//        new Thread().sleep(2000);
//        log.warn("异步修改结束");

        //同步修改节点
//        zkServer.getZooKeeper().delete("/testnode",2);

        //异步删除节点
//        String ctx = "{'delete':'success'}";
//        zkServer.getZooKeeper().delete("/testnode",2,new DeleteCallback(),ctx);
//        new Thread().sleep(2000);
//        log.warn("异步删除结束");

        //同步获取节点信息
//        Stat stat = new Stat();
//        byte[] resByte = zkServer.getZooKeeper().getData("/testnode",true,stat);
//        log.warn("同步获取节点信息,当前值："+new String(resByte)+",数据版本："+stat.getVersion());
//
//        //异步获取节点信息
//        String ctx = "{'get':'success'}";
//        zkServer.getZooKeeper().getData("/testnode",true,new DataCallback(),ctx);

        //同步获取子节点信息
//        List<String> listChildren = zkServer.getZooKeeper().getChildren("/testnode",true);
//        log.warn("同步获取子节点信息,子节点数量："+listChildren.size());


        //异步获取子节点信息
//        String ctx = "{'getChildren':'success'}";
//        zkServer.getZooKeeper().getChildren("/testnode",true,new ChildrenCallback(),ctx);
//        zkServer.getZooKeeper().getChildren("/testnode",true,new Children2Callback(),ctx);

        //判断节点是否存在
        //同步
//        Stat stat = zkServer.getZooKeeper().exists("/testnode",true);
//        if(null != stat)
//            log.warn("同步判断节点是否存在："+stat.toString());
//        else
//            log.warn("同步判断节点是否存在：不存在");
//        //异步
//        String ctx = "{'exists':'success'}";
//        zkServer.getZooKeeper().exists("/testnode",true,new StatCallback(),ctx);

        //自定义认证节点创建
//        List<ACL> listAcls = new ArrayList<ACL>();
//        Id id1 = new Id("digest",AclUtils.getDigestUserPwd("admin1:admin"));
//        Id id2 = new Id("digest",AclUtils.getDigestUserPwd("admin2:admin"));
//        listAcls.add(new ACL(ZooDefs.Perms.ALL,id1));
//        listAcls.add(new ACL(ZooDefs.Perms.READ,id2));
//        listAcls.add(new ACL(ZooDefs.Perms.DELETE | ZooDefs.Perms.CREATE,id2));
//        zkServer.zkNodeCreate("/acltest","acltest".getBytes(),listAcls);
        //执行完之后，可通过getAcl /acltest查看acl信息

        //注册过的用户必须通过addAuthInfo才能操作节点，参考命令行：addauth
//        zkServer.getZooKeeper().setData("/acltest","update".getBytes(),1);
//        zkServer.getZooKeeper().addAuthInfo("digest","admin1:admin".getBytes());
//        zkServer.getZooKeeper().setData("/acltest","update1".getBytes(),0);
//        log.warn("acltest修改完成");
//        zkServer.getZooKeeper().addAuthInfo("digest","admin2:admin".getBytes());
//        zkServer.getZooKeeper().setData("/acltest","update".getBytes(),0);

        //ip方式的acl
//        List<ACL> aclIps = new ArrayList<ACL>();
//        Id ipAcl1 = new Id("ip","127.0.0.1");
//        aclIps.add(new ACL(ZooDefs.Perms.ALL,ipAcl1));
//        zkServer.zkNodeCreate("/iptestlocal","iptest".getBytes(),aclIps);
//        zkServer.getZooKeeper().setData("/iptestlocal","update iptest".getBytes(),0);

        //此处设置可保证后台等待
        countDownLatch.await();
    }

    public void zkNodeCreate(String path ,byte[] data,List<ACL> acls){

        try {
            //同步创建节点
//            String result = zooKeeper.create(path+"tmp",data,acls, CreateMode.EPHEMERAL);
//            log.warn("同步创建节点：\t"+result +"\t成功");

            //异步创建节点
            String ctx = "{'create':'success'}";
            zooKeeper.create(path,data,acls, CreateMode.PERSISTENT,new CreateCallback(),ctx);
            new Thread().sleep(2000);
            log.warn("异步创建节点成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void process(WatchedEvent watchedEvent) {
        log.warn("接收到watch通知类型："+watchedEvent.getType());

        if (Event.EventType.NodeDataChanged == watchedEvent.getType()){
            countDownLatch.countDown(); //操作计数器，可通过命令行修改数据
        }else if (Event.EventType.NodeCreated == watchedEvent.getType()){

        }else if (Event.EventType.NodeDeleted == watchedEvent.getType()){

        }else if (Event.EventType.NodeChildrenChanged == watchedEvent.getType()){

        }else{

        }
    }
}
